/**
 * 
 */
package org.perfect.batch;

import java.util.concurrent.ThreadPoolExecutor;

import org.perfect.batch.launcher.ScheduledJobLauncher;
import org.perfect.batch.partition.DataSourcePartitioner;
import org.perfect.batch.partition.RemotePartitionHandler;
import org.perfect.batch.partition.RemoteStepExecutionAggregator;
import org.perfect.batch.step.PartitionMasterStep;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.partition.support.DefaultStepExecutionAggregator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

/**
 * @author billmse
 *
 */
@Configuration
@ConditionalOnClass({ BatchProperties.class })
@EnableConfigurationProperties({ BatchProperties.class })
public class BatchAutoConfiguration {

	@Autowired
	private BatchProperties batchProperties;
	
	private Logger logger = LoggerFactory.getLogger(getClass());

	@Bean(destroyMethod = "shutdown")
	public TaskScheduler customizedTaskScheduler() {
		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
		int poolSize = batchProperties.getScheduler().getPoolSize();
		taskScheduler.setPoolSize(poolSize);
		taskScheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		taskScheduler.initialize();
		if (logger.isInfoEnabled()) {
			logger.info("Initialize ThreadPoolTaskScheduler [poolSize = {}] successfully", poolSize);
		}
		return taskScheduler;
	}
	
	@Bean(destroyMethod = "shutdown")
	public TaskExecutor customizedTaskExecutor() {
		ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
		int corePoolSize = batchProperties.getExecutor().getCorePoolSize();
		taskExecutor.setCorePoolSize(corePoolSize);
		taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		taskExecutor.initialize();
		if (logger.isInfoEnabled()) {
			logger.info("Initialize ThreadPoolTaskExecutor [corePoolSize = {}] successfully", corePoolSize);
		}
		return taskExecutor;
	}

	@Bean
	@Primary
	public JobLauncher scheduledJobLauncher() {
		return new ScheduledJobLauncher();
	}

	@Bean
	public PartitionMasterStep partitionMasterStep() {
		return new PartitionMasterStep();
	}

	@Bean
	public RemotePartitionHandler remotePartitionHandler() {
		return new RemotePartitionHandler();
	}
	
	@Bean
	public DefaultStepExecutionAggregator defaultStepExecutionAggregator() {
		return new DefaultStepExecutionAggregator();
	}

	@Bean
	public RemoteStepExecutionAggregator remoteStepExecutionAggregator() {
		return new RemoteStepExecutionAggregator();
	}
	
	@Bean
	public DataSourcePartitioner dataSourcePartitioner() {
		return new DataSourcePartitioner();
	}

}
